// Tencent is pleased to support the open source community by making
// 蓝鲸智云 - 监控平台/日志平台 (BlueKing - Monitor/Log) available.
// Copyright (C) 2017-2022 THL A29 Limited, a Tencent company. All rights reserved.
// Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
// You may obtain a copy of the License at http://opensource.org/licenses/MIT
// Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
// an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
// specific language governing permissions and limitations under the License.
//

package exporter

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/TencentBlueKing/bkmonitor-kits/logger"
	"github.com/TencentBlueKing/collector-go-sdk/v2/bkbeat/beat"

	"github.com/TencentBlueKing/bk-collector/confengine"
	"github.com/TencentBlueKing/bk-collector/define"
	"github.com/TencentBlueKing/bk-collector/exporter/converter"
	"github.com/TencentBlueKing/bk-collector/exporter/queue"
	"github.com/TencentBlueKing/bk-collector/internal/monitoring"
)

const (
	ConfigFieldExporter = "exporter"
)

type Exporter struct {
	ctx       context.Context
	cancel    context.CancelFunc
	wg        sync.WaitGroup
	converter converter.Converter
	config    Config
	queue     queue.Queue
}

var globalRecords = define.NewRecordQueue(define.PushModeGuarantee)

func PublishRecord(r *define.Record) {
	globalRecords.Push(r)
}

func New(conf *confengine.Config) (*Exporter, error) {
	c := Config{}
	if err := conf.UnpackChild(ConfigFieldExporter, &c); err != nil {
		return nil, err
	}
	logger.Infof("Exporter config: %+v", c)

	ctx, cancel := context.WithCancel(context.Background())
	return &Exporter{
		ctx:       ctx,
		cancel:    cancel,
		config:    c,
		converter: converter.NewCommonConverter(),
		queue:     queue.NewBatchQueue(c.Queue.BatchSize, c.Queue.FlushInterval),
	}, nil
}

func (e *Exporter) Start() error {
	logger.Infof("Exporter start working...")
	for i := 0; i < define.CoreNums(); i++ {
		go e.consumeRecords()
		go e.sendEvents()
	}
	return nil
}

func (e *Exporter) consumeRecords() {
	e.wg.Add(1)
	defer e.wg.Done()

	for {
		select {
		case record := <-globalRecords.Get():
			for _, evt := range e.converter.Convert(record) {
				lvs := []string{string(record.RecordType), fmt.Sprintf("%d", record.Token.Get(record.RecordType))}
				monitoring.ExporterSentTotal.WithLabelValues(lvs...).Inc()
				e.queue.Put(evt)
			}

		case <-e.ctx.Done():
			return
		}
	}
}

func (e *Exporter) sendEvents() {
	e.wg.Add(1)
	defer e.wg.Done()

	for {
		select {
		case evt := <-e.queue.Pop():
			start := time.Now()
			beat.Send(evt)
			monitoring.ExporterSentDuration.WithLabelValues().Observe(float64(time.Since(start).Milliseconds()))

		case <-e.ctx.Done():
			e.queue.Close()
			return
		}
	}
}

func (e *Exporter) Stop() error {
	e.cancel()
	e.wg.Wait()
	return nil
}
